package com.wen.service.impl;

import com.wen.entity.po.AppUser;
import com.wen.mappers.AppUserMapper;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Service;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.stream.Collectors;
import java.util.stream.IntStream;

@Service
public class UserBatchService {

    @Autowired
    private AppUserMapper appUserMapper;

    @Autowired
    private ThreadPoolTaskExecutor taskExecutor;

    /**
     * 并行查询100万条数据（分10个线程，每个线程查10万条）
     */
    public List<AppUser> query100wUsers() {
        // 计算总页数和每页大小
        int totalData = 1000000;
        int pageSize = 100000;
        int totalPages = (totalData + pageSize - 1) / pageSize; // 10页

        // 创建所有并行查询任务
        List<CompletableFuture<List<AppUser>>> futures = IntStream.range(0, totalPages)
                .mapToObj(pageNum -> createQueryTask(pageNum, pageSize))
                .collect(Collectors.toList());

        // 等待所有任务完成并汇总结果
        CompletableFuture<Void> allFutures = CompletableFuture.allOf(
                futures.toArray(new CompletableFuture[0])
        );
//        CompletableFuture[] cfs =(CompletableFuture[]) futures.toArray();
        // 收集所有结果
//        allFutures.join();
//        return allFutures.thenApply(v ->
//                futures.stream()
//                        .map(CompletableFuture::join)
//                        .flatMap(List::stream)
//                        .collect(Collectors.toList())
//        ).join();
        // 1. 先等所有任务都完成（allFutures就是那个"等待所有任务"的触发器）
        allFutures.join();

        // 2. 创建一个空列表，用来装所有结果
        List<AppUser> allResults = new ArrayList<>();

        // 3. 遍历每个异步任务
        for (CompletableFuture<List<AppUser>> future : futures) {
            // 3.1 获取这个任务的结果（因为前面已经等所有任务完成了，这里不会卡）
            List<AppUser> singleResult = future.join();

            // 3.2 把这个任务的结果加到总列表里
            allResults.addAll(singleResult);
        }

        // 4. 返回合并后的总列表
        return allResults;
    }

    /**
     * 创建单个查询任务
     */
    private CompletableFuture<List<AppUser>> createQueryTask(int pageNum, int pageSize) {
        // 使用自定义线程池执行任务
        return CompletableFuture.supplyAsync(() -> {
            try {
                int offset = pageNum * pageSize;
                String threadName = Thread.currentThread().getName();
                System.out.printf("线程 %s 开始查询: 偏移量=%d, 数量=%d%n", threadName, offset, pageSize);

                // 执行分页查询
                List<AppUser> users = appUserMapper.selectByPage(offset, pageSize);

                System.out.printf("线程 %s 完成查询: 已获取 %d 条数据%n", threadName, users.size());
                return users;
            } catch (Exception e) {
                System.err.printf("查询失败 (pageNum=%d): %s%n", pageNum, e.getMessage());
                return new ArrayList<>(); // 出错时返回空列表，避免影响整体结果
            }
        }, taskExecutor);
    }


}
